package rocketmq.manager;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import rocketmq.domain.RocketMqMessage;

import javax.annotation.Resource;

/**
 * @author mercury
 */
@Component
@Slf4j
public class RocketMqManager {

    @Resource
    private RocketMQTemplate rocketMqTemplate;


    private static final int SEND_MSG_TIMEOUT = 3000;


    /**
     * 同步发送消息
     *
     * @param topic           主题
     * @param rocketMqMessage 内容
     */
    public void syncSend(String topic, RocketMqMessage rocketMqMessage) {
        doSend(topic, SEND_MSG_TIMEOUT, 0, rocketMqMessage);
    }


    /**
     * 同步发送延时消息
     *
     * @param topic           主题
     * @param delayLevel      延时级别
     * @param rocketMqMessage 内容
     */
    public void syncDelaySend(String topic, Integer delayLevel, RocketMqMessage rocketMqMessage) {
        doSend(topic, SEND_MSG_TIMEOUT, delayLevel, rocketMqMessage);
    }


    /**
     * 发送消息
     *
     * @param topic           主题
     * @param timeout         超时时间
     * @param delayLevel      延时级别
     * @param rocketMqMessage 消息内容
     */
    public void doSend(String topic, Integer timeout, Integer delayLevel, RocketMqMessage rocketMqMessage) {
        Message<RocketMqMessage> message = MessageBuilder.withPayload(rocketMqMessage).build();
        rocketMqTemplate.syncSend(topic, message, timeout, delayLevel);
    }


    /**
     * 异步发送消息
     *
     * @param topic           主题
     * @param rocketMqMessage 消息内容
     */
    public void sendAsyncMsg(String topic, RocketMqMessage rocketMqMessage) {
        Message<RocketMqMessage> message = MessageBuilder.withPayload(rocketMqMessage).build();
        rocketMqTemplate.asyncSend(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                    log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());
                } else {
                    log.debug("Consumer replies message success.");
                }
            }

            @Override
            public void onException(Throwable e) {
                log.error("Consumer replies message failed. error: {}", e.getLocalizedMessage());
            }
        });
    }

}
